package com.binance.zkconfig.app.util;

import com.binance.zkconfig.app.constant.Constant;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.KeeperException;
import org.springframework.util.ReflectionUtils;

import java.nio.charset.Charset;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by Fei.Huang on 2018/8/28.
 */
@Slf4j
public final class ZkConfigLoader {

    public static Map<String, String> loadProperties(String context, CuratorFramework curator) {
        Map<String, String> properties = new LinkedHashMap<>();
        findProperties(context, curator, context, null, properties);
        return properties;
    }

    private static byte[] getPropertyBytes(CuratorFramework curator, String nodePath)
            throws KeeperException {
        try {
            return curator.getData().forPath(nodePath);
        } catch (KeeperException e) {
            if (e.code() != KeeperException.Code.NONODE) {
                throw e;
            }
        } catch (Exception e) {
            log.error("ZkConfig getPropertyValue failed, nodePath:{}", nodePath, e);
            ReflectionUtils.rethrowRuntimeException(e);
        }
        return null;
    }

    private static void findProperties(String context, CuratorFramework curator, String nodePath,
                                       List<String> children, Map<String, String> source) {
        try {
            log.debug("ZkConfig findProperties start, nodePath:{}", nodePath);
            List<String> finalChildren = children == null ? getChildren(curator, nodePath) : children;
            if (finalChildren == null || finalChildren.isEmpty()) {
                return;
            }
            for (String child : finalChildren) {
                String childPath = nodePath + Constant.SLASH + child;
                List<String> childPathChildren = getChildren(curator, childPath);

                byte[] bytes = getPropertyBytes(curator, childPath);
                if (bytes == null || bytes.length == 0) {
                    if (childPathChildren == null || childPathChildren.isEmpty()) {
                        registerKeyValue(context, source, childPath, "");
                    }
                } else {
                    registerKeyValue(context, source, childPath, new String(bytes, Charset.forName(Constant.UTF_8)));
                }

                findProperties(context, curator, childPath, childPathChildren, source);
            }
            log.debug("ZkConfig findProperties end, nodePath:{}", nodePath);
        } catch (Exception e) {
            log.error("ZkConfig findProperties failed, nodePath:{}", nodePath, e);
            ReflectionUtils.rethrowRuntimeException(e);
        }
    }

    private static List<String> getChildren(CuratorFramework curator, String path) throws Exception {
        List<String> children = null;
        try {
            children = curator.getChildren().forPath(path);
        } catch (KeeperException e) {
            if (e.code() != KeeperException.Code.NONODE) {
                throw e;
            }
        }
        return children;
    }

    private static void registerKeyValue(String context, Map<String, String> source, String path,
                                         String value) {
        String key = ZkConfigUtils.sanitizeKey(context, path);
        source.put(key, value);
    }
}
